package com.we.risk.phoneRecord.var;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.doris.SinkDoris;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.phoneRecord.stage2.adm.AdmUserLabel;
import com.we.utils.ParamCheck;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.Properties;

@Deprecated
public class AtoCallRecordTest {
    public static final String RELEASEPROP =
            "risk/phonerecord/var/adm_user_label_kfk_self_test.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmUserLabel.class);

    public static final String INPUTKEYBY = "user_key";
    public static final String RECVTIME = "recv_time";
    public static final String OUTKEYBY = "user_key";
    public static final String LABELOUTIME = "label_output_time";
    public static final String OUTJOBID = "job_id";
    public static final String JOBID = "jobID";
    public static final String PHONECALLLIST = "phoneCallList";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 1000); // ms
            ckConf.setCheckpointTimeout(5* 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "PhoneRecord")
                            .uid("PhoneRecord-var-source");


            SingleOutputStreamOperator<String> out = input.filter(
                    new FilterFunction<String>() {
                        @Override
                        public boolean filter(String value) throws Exception {
                            JSONObject inputJson = JSONObject.parseObject(value);
                            if (inputJson.containsKey(INPUTKEYBY)
                                    && inputJson.getString(INPUTKEYBY) != null) {
                                return true;
                            }
                            return false;
                        }
                    })
                    .keyBy(
                            new KeySelector<String, String>() {
                                @Override
                                public String getKey(String value) throws Exception {
                                    JSONObject inputJson = JSONObject.parseObject(value);
                                    return inputJson.getString(INPUTKEYBY);
                                }
                            })
                    .flatMap(new UserLabelRichFlatMapFunc());

            DorisSink<String> dorisSink = SinkDoris.buidJsonArrayDorisSink("10.10.148.68:8030,10.10.148.69:8030,10.10.148.74:8030",
                    "ato_var_callrecord.var_230630",
                    "cr_var_230630",
                    "doris",
                    "wetc@#dw23",
                    "recv_time");

            out.sinkTo(dorisSink).setParallelism(4);

            env.execute(AtoCallRecordTest.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class UserLabelRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> dimUserVState;
        ValueState<Integer> callListSizeVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            dimUserVState =
                getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("dimUserVState", JSONObject.class));
            callListSizeVState =
                getRuntimeContext().getState(new ValueStateDescriptor<Integer>("callListSize", Integer.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                JSONObject tmpUserState = dimUserVState.value();
                JSONObject curUserState = (tmpUserState == null ? new JSONObject() : tmpUserState);

                if (value.contains(JOBID)) {
                    //phoneCalllist
                    LOG.debug(
                            "Debug curUserState.size: "
                                    + curUserState.size()
                                    + " curstate: "
                                    + curUserState.toString());
                    if (curUserState.size() > 1) {
                        JSONArray res = ProcessData(value, curUserState, callListSizeVState);
                        out.collect(res.toString());
                    }
                } else {
                    //userInfo
                    LOG.debug("Debug: dim UserKey come in: " + value);
                    JSONObject inputJson = JSONObject.parseObject(value);
                    if (!curUserState.equals(inputJson)) {
                        dimUserVState.update(inputJson);
                    }
                }
            } catch (Exception e) {
                LOG.error("UserLabelRichFlatMapFunc Exception: " + e.toString());
                LOG.error(
                        "UserLabelRichFlatMapFunc Exception input: "
                                + (value.length() >= 100 ? value.substring(0, 100) : value));
                e.printStackTrace();
            }
        }

        private JSONArray ProcessData(String value, JSONObject userState, ValueState<Integer> callListSizeVState) throws Exception {
            JSONArray resJsonArray = new JSONArray();

            Integer tmpCallListSize = callListSizeVState.value();
            int lastCallListSize = (tmpCallListSize == null ? 0 : tmpCallListSize);

            JSONObject inputJson = JSONObject.parseObject(value);
            JSONObject commonJson = new JSONObject();
            commonJson.put(OUTKEYBY, inputJson.getString(INPUTKEYBY));
            commonJson.put(OUTJOBID, inputJson.getString(JOBID));
            commonJson.put(RECVTIME, inputJson.getLong(RECVTIME));

            JSONArray phoneCallList = inputJson.getJSONArray(PHONECALLLIST);
            ParamCheck.isNull(phoneCallList, "phoneCallList");
            int size = phoneCallList.size();

            for (int i = 0; i < size; i++) {
                JSONObject callDetail = phoneCallList.getJSONObject(i);
                callDetail.putAll(commonJson);
                callDetail.put("row_num", i);
                callDetail.put("is_active", 1);

                formatSamePositions(callDetail, userState);
                resJsonArray.add(callDetail);
            }

            if(size < lastCallListSize) {
                for (int i = size; i < lastCallListSize; i++) {
                    JSONObject expiredCall = new JSONObject();
                    expiredCall.putAll(commonJson);
                    expiredCall.put("row_num", i);
                    expiredCall.put("is_active", 0);

                    resJsonArray.add(expiredCall);
                }
                callListSizeVState.update(size);
            }

            return resJsonArray;
        }

        private void formatSamePositions(JSONObject callDetail, JSONObject userState) {
            byte samePidProv = 0;
            byte samePidCity = 0;
            byte sameMobProv = 0;
            byte sameMobCity = 0;
            byte sameAddProv = 0;
            byte sameAddCity = 0;
            byte sameMobOpt = 0;

            if (callDetail.containsKey("mobile_prov")) {
                String mob_prov = callDetail.getString("mobile_prov");
                String state_mob_prov = userState.getString("mob_prov");
                String state_pid_prov = userState.getString("pid_prov");
                String state_ad_prov = userState.getString("ad_prov");

                sameMobProv = (byte) (mob_prov.equals(state_mob_prov) ? 1 : 0);
                samePidProv = (byte) (mob_prov.equals(state_pid_prov) ? 1 : 0);
                sameAddProv = (byte) (mob_prov.equals(state_ad_prov) ? 1 : 0);
            }

            if (callDetail.containsKey("mobile_city")) {
                String mob_city = callDetail.getString("mobile_city");
                String state_mob_city = userState.getString("mob_city");
                String state_pid_city = userState.getString("pid_city");
                String state_ad_city = userState.getString("ad_city");

                sameMobCity = (byte) (mob_city.equals(state_mob_city) ? 1 : 0);
                samePidCity = (byte) (mob_city.equals(state_pid_city) ? 1 : 0);
                sameAddCity = (byte) (mob_city.equals(state_ad_city) ? 1 : 0);
            }

            if (callDetail.containsKey("mobile_opt")) {
                String mob_opt = callDetail.getString("mobile_opt");
                String state_mob_isp = userState.getString("mob_isp");
                sameMobOpt = (byte) (mob_opt.equals(state_mob_isp) ? 1: 0);
            }

            callDetail.put("same_positions", ""+samePidProv+samePidCity+sameMobProv+sameMobCity+sameMobOpt+sameAddProv+sameAddCity);
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTKEYBY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }


}
